package com.youxin.busfoundation.common.es;

import java.util.concurrent.TimeUnit;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import com.youxin.busfoundation.config.EsConfig;

/**
 * @desc ES客户端操作类
 * @author yingde.cao
 * @update 2017-12-18
 */
public class EsClient {
	private static EsClient esClient = null;
	private EsConfig config;
	private Client client = null;

	public EsClient(EsConfig config) {
		this.config = config;		
	}

	public static EsClient instance() {
		if (esClient != null) {
			return esClient;
		}
		synchronized (EsClient.class) {
			if (esClient == null) {
				esClient = new EsClient(EsConfig.getConfig());
			}
		}
		return esClient;
	}

	public Client getClient() throws Exception{
		initClient();		
		return client;
	}

	/**
	 * @desc 执行ES
	 * @param statement
	 * @throws Exception
	 */
	public <T> T execute(EsExecuteStatement<T> statement) throws Exception {
		try {
			initClient();
			return statement.execute(client);
		} catch (Exception e) {
			throw e;
		} finally {}
	}
	
	public synchronized void closeClient()
	{
		try {
			if (client != null) {
				client.close();
			}
		} catch (Exception e) {}
	}

	public static void shutdown(){
		esClient.closeClient();
	}
	
	/**
	 * @desc 初始化方法
	 * @return
	 * @throws Exception
	 */
	@SuppressWarnings("resource")
	private void initClient() throws Exception {
		if (client != null) {
			return;
		}
		synchronized (this) {
			if (client != null) {
				return;
			}
			TransportAddress[] hostAndPorts = config.getHostAndPorts();
			if (hostAndPorts == null || hostAndPorts.length == 0) {
				throw new IllegalArgumentException("Elasticsearch主机配置有误");
			}
			PreBuiltTransportClient tclient = null;
			Settings settings = Settings.builder()
					// .put("cluster.name", "sls_es_test")
					.put("client.transport.ignore_cluster_name", true).put("client.transport.sniff", false)
					.put("client.transport.ping_timeout", 20, TimeUnit.SECONDS).build();
			tclient = new PreBuiltTransportClient(settings);
			tclient.addTransportAddresses(hostAndPorts);
			this.client = tclient;
		}
	}

}
